package com.flume.sink;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import javax.print.DocFlavor.BYTE_ARRAY;

import org.apache.commons.lang.NullArgumentException;
import org.apache.flume.Channel;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Transaction;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;
import org.bson.Document;
import org.joda.time.format.DateTimeFormat;
import org.joda.time.format.DateTimeFormatter;
import org.joda.time.format.DateTimeFormatterBuilder;
import org.joda.time.format.DateTimeParser;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.mongodb.DBObject;
import com.mongodb.MongoClient;
import com.mongodb.ServerAddress;
import com.mongodb.client.MongoDatabase;
import com.mongodb.util.JSON;

/**
 * 
 * MongoSink  
 * <p>将获取的日志存储到mongodb中
 * 目前只支持同库不同集合。</p> 
 * @author sunChangXing
 * @version v1.0.0
 * <p><B>last update </B> by sunChangXing @ 2017年12月25日</p>
 * @since v1.0.0
 */
public class MongoSink extends AbstractSink implements Configurable {
	private Logger logger = LoggerFactory.getLogger(MongoSink.class);
	/**mongodb数据库*/
	private MongoDatabase mongoDatabase = null;

	/**
	 * 配置时间转码器（感觉不错，虽然在目前没有用到）
	 */
	private DateTimeParser[] parsers = { DateTimeFormat.forPattern("yyyy-MM-dd").getParser(),
			DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss").getParser(),
			DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss.SSS").getParser(),
			DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss Z").getParser(),
			DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss.SSS Z").getParser(),
			DateTimeFormat.forPattern("yyyy-MM-dd'T'HH:mm:ssZ").getParser(),
			DateTimeFormat.forPattern("yyyy-MM-dd'T'HH:mm:ss.SSSZ").getParser(),
			DateTimeFormat.forPattern("yyyy-MM-dd'T'HH:mm:ssz").getParser(),
			DateTimeFormat.forPattern("yyyy-MM-dd'T'HH:mm:ss.SSSz").getParser()};
	
	private DateTimeFormatter dateTimeFormatter = new DateTimeFormatterBuilder().append(null, parsers).toFormatter();
	
	public static void main(String[] args) {
//		List<ServerAddress> seeds = new ArrayList<ServerAddress>();
//		ServerAddress serverAddress = new ServerAddress("127.0.0.1", 27017);
//		seeds.add(serverAddress);
//		mongoDatabase = new MongoClient(seeds).getDatabase("test");
//		Document saveDoc = new Document();
//		saveDoc.append("$set", new Document().append("b", 0));
//		Document searchDoc = new Document();
//		searchDoc.append("_id", new ObjectId("5a0e91ff4d78d6715be15794"));
//		UpdateResult updateresult = mongoDatabase.getCollection("mm").updateOne(searchDoc, saveDoc);
//		System.out.println(updateresult.getModifiedCount());
	}
	/**
	 * 外部暴露的参数的key
	 */
	/**host*/ 
	private final String HOST = "host";
	/**port*/
	private final String PORT = "port";
	/**链接的数据库名*/
	private final String DATABASE = "dataBase";
	/**数据库的日志信息集合*/
	private final String LOG_COLLECTION = "defaultColltion";
	/**批处理的量*/
	private final String BATCH_SIZE = "batchSize";
	/**保存默认集合*/
	private String DEFAULT_COLLECTION;
	/**批量获取event的大小*/
	private int batchSize;


	/**初始化mongodb日志数据*/
	@Override
	public void configure(Context context) {
		try {
			String host = context.getString(HOST);
			if (null == host) {
				throw new NullArgumentException("host is null");
			}
			Integer post = context.getInteger(PORT, 27017);
			if (null == post) {
				throw new NullArgumentException("post is null");
			}
			String dataBase = context.getString(DATABASE);
			if (null == dataBase) {
				throw new NullArgumentException("dataBase is null");
			}
			DEFAULT_COLLECTION = context.getString(LOG_COLLECTION);
			if (null == DEFAULT_COLLECTION) {
				throw new NullArgumentException("defaultCollection is null");
			}
			batchSize = context.getInteger(BATCH_SIZE, 100);
			// 配置mongodb链接
			List<ServerAddress> seeds = new ArrayList<ServerAddress>();
			ServerAddress serverAddress = new ServerAddress(host, post);
			seeds.add(serverAddress);
			mongoDatabase = new MongoClient(seeds).getDatabase(dataBase);
		} catch (Exception e) {
			logger.error("初始化mongodb配置失败" + e);
		}
	}

	/**
	 * 执行日志数据的插入处理
	 */
	@Override
	public Status process() throws EventDeliveryException {
		// 设置状态恢复
		Status status = null;
		// 获取通道信息
		Channel ch = getChannel();
		// 开启事物
		Transaction transaction = null;
		transaction = ch.getTransaction();
		// 执行批量插入的集合
		Map<String, LinkedList<Document>> upsertMap = new HashMap<String, LinkedList<Document>>();
		try {
			// 处理集合里面的文档对象是否符合批处理的
			transaction.begin();
			Event event = null;
			// 设置文档批处理
			for (int i = 0; i < batchSize; i++) {
				event = ch.take();
				if (event == null) {
					break;
				} 
				logger.info("----------" + new String(event.getBody()));
				// 执行转换event
				parseEvent(upsertMap, event);
			}
			if (!upsertMap.isEmpty()) {
				// 插入当前的数据
				batchInsertLog(upsertMap);
			}
			// 标记状态已读
			status = Status.READY;
			// 成功提交清空通道信息
			transaction.commit();
		} catch (Exception e) {
			transaction.rollback();
			status = Status.BACKOFF;
			logger.info(e.getMessage());
		} finally {
			transaction.close();
		}
		return status;
	}
	
	/**
	 * 
	 * 分集合插入。
	 * <p></p>
	 * @param upsertMap 数据对象
	 */
	private void batchInsertLog(Map<String, LinkedList<Document>> upsertMap) {
		Iterator<String> keySets =  upsertMap.keySet().iterator();
		while(keySets.hasNext()) {
			// 逐集合的插入
			String collection = keySets.next();
			LinkedList<Document> addDocs = upsertMap.get(collection);
			if (null != addDocs && !addDocs.isEmpty()) {
				mongoDatabase.getCollection(collection).insertMany(addDocs);
			}
		}
	}

	/**
	 * 
	 * event处理。
	 * <p></p>
	 * @param upsertMap 插入对象
	 * @param event 对象
	 * @throws Exception
	 */
	private void parseEvent(Map<String, LinkedList<Document>> upsertMap, Event event) throws Exception {
		// 处理头部信息
		handleHeader();
		// 处理body主体信息
		handleBody(upsertMap, event.getBody());
	}
	
	/**
	 * 
	 * 处理请求头。
	 * <p></p>
	 */
	private void handleHeader() {
		
	} 
	
	/**
	 * 
	 * 转换body内容。
	 * <p></p>
	 * @param bodys  默认2048个字节，在LineDeserializer中设置长度
	 */
	private void handleBody(Map<String, LinkedList<Document>> upsertMap, byte[] bodys) throws Exception {
		// 判断bodys是否为JSONObject类型
		DBObject bodysJson = (DBObject) JSON.parse(new String(bodys, "UTF-8"));
		// 判断来源项目名
		Object projectName = bodysJson.get("projectName");
		if (null == projectName) {
			throw new NullArgumentException("body key projectName is null");
		}
		Object collectionName = bodysJson.get("collection");
		String collection = DEFAULT_COLLECTION;
		if (null != collectionName && !"".equals(collectionName)) {
			collection = String.valueOf(collectionName);
		}
		// 判断当前是否存在该集合名
		if (!upsertMap.containsKey(collection)) {
			upsertMap.put(collection, new LinkedList<Document>());
		}
		List<Document> docs = upsertMap.get(collection);
		if (!bodysJson.containsField("data")) {
			throw new Exception("body key data is null");
		}
		// 实体日志数据
		DBObject dataJson = (DBObject) bodysJson.get("data");
		Set<String> dataSet = dataJson.keySet();
		Iterator<String> dataIterator = dataSet.iterator();
		Document addDoc = new Document();
		while (dataIterator.hasNext()) {
			String dataKey= dataIterator.next();
			addDoc.append(dataKey, dataJson.get(dataKey));
		}
		docs.add(addDoc);
	}
	
}
